// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/korbit/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/korbit/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::korbit::feed_v1::impl {

void KorbitFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError());
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document["data"].HasMember("channel")) {
    const std::string& channel = document["data"]["channel"].GetString();

    if (!CheckChannelFilter(channel)) {
      return;
    }

    if (channel == "transaction") {
      const auto& transaction_data = document["data"];
      const auto& native_symbol = transaction_data["currency_pair"].GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }
      Trade* trade = upd_mgr_.GetTradeMutable();
      KorbitFeedUpdateParser::ParseTrade(transaction_data, trade);
      upd_mgr_.PublishTrade(native_symbol, false);
    } else if (channel == "orderbook") {
      const auto& orderbook_data = document["data"];
      std::string native_symbol = orderbook_data["currency_pair"].GetString();
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        KorbitFeedUpdateParser::ParseSnapshot(book, orderbook_data);
        upd_mgr_.PublishBook();
      }
    } else {
      LOG(INFO) << "[Korbit Feed Processor] unknown msg for " << packet.topic_name << ": "
                << packet.payload;
    }
  } else if (document.HasMember("url")) {
    const std::string& url = document["url"].GetString();
    std::vector<std::string> splited = Split(url, "?");
    CHECK_EQ(splited.size(), 2);
    if (StringEndsWith(splited[0], "/v1/candles/tradingview")) {
      std::string channel = "kline_rest";
      if (!CheckChannelFilter(channel)) {
        return;
      }
      auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      const std::string& native_symbol = parameters["symbol"];
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }

      auto interval = KorbitFeedUpdateParser::ParseKlineInterval(parameters["resolution"]);
      auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (klines != nullptr) {
        auto data = document["data"]["data"].GetArray();
        std::optional<int64_t> first_open_timestamp;
        if (data.Size() > 0) {
          first_open_timestamp = stoll(data[0]["t"].GetString()) * 1'000'000LL;
        }

        if (klines->Size() == 0) {
          klines->SnapshotClear();
        } else if (first_open_timestamp &&
                  first_open_timestamp < klines->RBegin()->open_timestamp) {
          klines->SnapshotClear();
        } else {
          klines->UpdateClear();
        }
        bool to_publish = false;
        for (const auto& kline_data : data) {
          bool freezed = KorbitFeedUpdateParser::ParseKlineRest(kline_data,
              klines, interval, packet.timestamp);
          to_publish = to_publish || freezed;
        }
        if (to_publish) upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else {
    // meta msg like subscription response...
    LOG(INFO) << "[Korbit Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload;
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::korbit::feed_v1::impl
